insert overwrite table jms_dm.dm_sqs_false_sign_warning_mark_detail_hi partition (dt)
select bill_code,
       max(latest_scan_date)        latest_scan_date,
       max(latest_scan_code)        latest_scan_code,
       max(biz_sign_date)           biz_sign_date,
       max(biz_sign_site_code)      biz_sign_site_code,
       max(last_site_come_date)     last_site_come_date,
       max(last_site_come_name)     last_site_come_name,
       max(last_site_come_code)     last_site_come_code,
       max(collect_in_site_date)    collect_in_site_date,
       max(collect_in_site_name)    collect_in_site_name,
       max(collect_in_site_code)    collect_in_site_code,
       max(collect_out_site_date)   collect_out_site_date,
       max(collect_out_site_name)   collect_out_site_name,
       max(collect_out_site_code)   collect_out_site_code,
       max(receiver_detail_address) receiver_detail_address,
       max(problem_desc_code)       problem_desc_code,
       max(problem_desc)            problem_desc,
       max(is_warning_keys)         is_warning_keys,
       max(is_out_violate)          is_out_violate,
       max(is_in_violate)           is_in_violate,
       max(is_in_call_violate_must) is_in_call_violate_must,
       max(is_in_call_violate)      is_in_call_violate,
       max(is_sign_call_violate)    is_sign_call_violate,
       max(call_time)               call_time,
       max(signed_type)             signed_type,
       max(create_time)             create_time,
       max(work_type)               work_type,
       max(partner_name)            partner_name,
       max(storage_address)         storage_address,
       max(is_problem)              is_problem,
       max(is_problem_warning)      is_problem_warning,
       max(is_problem_time)         is_problem_time,
       max(is_problem_call)         is_problem_call,
       max(new_content)             new_content,
       max(leave_time)              leave_time,
       max(is_plan)                 is_plan,
       max(is_leave)                is_leave,
       max(send_deliver_user_id)    send_deliver_user_id,
       max(send_deliver_user)       send_deliver_user,
       '{{ execution_date | cst_hour }}'                 dt
from (select *
      from (select detail.bill_code,
                   detail.latest_scan_date,
                   detail.latest_scan_code,
                   detail.biz_sign_date,
                   detail.biz_sign_site_code,
                   detail.last_site_come_date,
                   detail.last_site_come_name,
                   detail.last_site_come_code,
                   detail.collect_in_site_date,
                   detail.collect_in_site_name,
                   detail.collect_in_site_code,
                   detail.collect_out_site_date,
                   detail.collect_out_site_name,
                   detail.collect_out_site_code,
                   detail.receiver_detail_address,
                   detail.problem_desc_code,
                   detail.problem_desc,
                   if(receiver_detail_address REGEXP
                      '送货上门|不要放|不放驿站|不要代签收|拒收极兔|务必送|务必验货|不能放|务必打电话|拒绝驿站|请勿放|勿放|拒绝代收|别放|孕妇|否则投诉|老人|准放|派送上门|不允许放|私自签收|请勿放|请放|请电联',
                      1, 0)                                                      is_warning_keys,
                   if((hour(collect_out_site_date) >= 0 and hour(collect_out_site_date) < 6) and
                      collect_out_site_date is not null, 1, 0)                   is_out_violate,
                   if((unix_timestamp(collect_out_site_date) - unix_timestamp(collect_in_site_date)) < 10
                          and (unix_timestamp(collect_out_site_date) - unix_timestamp(collect_in_site_date)) > 0
                          and collect_out_site_date is not null and collect_in_site_date is not null, 1,
                      0)                                                         is_in_violate,
                   if(collect_in_site_date is not null and collect_out_site_date is null and biz_sign_date is null and
                      mark.is_contact = '0', 1, 0)                               is_in_call_violate_must,
                   if(collect_in_site_date is not null and collect_out_site_date is null and biz_sign_date is null and
                      mark.is_contact is null and call.waybill_no is null, 1, 0) is_in_call_violate,
                   if(biz_sign_date is not null and
                      (nvl(sign.signed_type, '') REGEXP '家人待收|门卫|物业|超时|收发室|代收点|前台')
                          and
                      call.waybill_no is null
                       , 1, 0)                                                   is_sign_call_violate,
                   call.start_time                                               call_time,
                   sign.signed_type,
                   complaint.create_time,
                   complaint.work_type,
                   collect.partner_name,
                   collect.storage_address,
                   if(problem_desc_code in ('15a', '14a' ,'65c' ,'65d'), 1, 0)                 is_problem,
                   if(problem_desc_code = '15a' and collect_out_site_date is null and biz_sign_date is null, 1,
                      0)                                                         is_problem_warning,
                   if(problem_desc_code = '14a' and collect_out_site_date is null and biz_sign_date is null, 1,
                      0)                                                         is_problem_time,
                   if(problem_desc_code in ('65c','65d') and collect_out_site_date is null and biz_sign_date is null, 1,
                      0)                                                         is_problem_call,
                   null                                                          new_content,
                   null                                                          leave_time,
                   0                                                             is_plan,
                   0                                                             is_leave,
                   nvl(detail.send_deliver_user_id,collect.send_deliver_userid) send_deliver_user_id,
                   nvl(detail.send_deliver_user,collect.send_deliver_user) send_deliver_user
            from (select bill_code
                       , if(deliver_date = '', null, deliver_date)                      latest_scan_date
                       , deliver_site_code                                              latest_scan_code
                       , if(biz_sign_date = '1999-01-01 00:00:00', null, biz_sign_date) biz_sign_date
                       , if(biz_sign_site_code = '', null, biz_sign_site_code)          biz_sign_site_code
                       , if(deliver_date = '1999-01-01 00:00:00', null, deliver_date)   last_site_come_date
                       , if(deliver_site_name = '', null, deliver_site_name)            last_site_come_name
                       , if(deliver_site_code = '', null, deliver_site_code)            last_site_come_code
                       , if(collect_in_site_date = '1999-01-01 00:00:00', null,
                            collect_in_site_date)                                       collect_in_site_date
                       , if(collect_in_site_name = '', null, collect_in_site_name)      collect_in_site_name
                       , if(collect_in_site_code = '', null, collect_in_site_code)      collect_in_site_code
                       , if(collect_out_site_date = '1999-01-01 00:00:00', null,
                            collect_out_site_date)                                      collect_out_site_date
                       , if(collect_out_site_name = '', null, collect_out_site_name)    collect_out_site_name
                       , if(collect_out_site_code = '', null, collect_out_site_code)    collect_out_site_code
                       , receiver_detail_address
                       , second_level_type_code
                       , receive_network_code
                       , receive_network_name
                       , if(problem_desc_code = '', null, problem_desc_code)            problem_desc_code
                       , if(problem_desc = '', null, problem_desc)                      problem_desc
                       , if(send_deliver_user_id='' or send_deliver_user_id=0,null,send_deliver_user_id) send_deliver_user_id
                       , if(send_deliver_user='',null,send_deliver_user) send_deliver_user
                  from jms_dwd.dwd_wide_yl_rt_ssmx_bill_detail_hi
                  where dt = '{{ execution_date | cst_hour }}'
                    and latest_scan_type > 6
                    and date(deliver_date) >= date_add('{{ execution_date | cst_ds }}', -7)
                    and date(deliver_date) <= '{{ execution_date | cst_ds }}'
                    and deliver_site_code <> '') detail
                     left join jms_dwd.dwd_call_record_hi call
                               on substr(call.dt, 1, 10) >= date_add('{{ execution_date | cst_ds }}', -6)
                                   and substr(call.dt, 1, 10) <= '{{ execution_date | cst_ds }}'
                                   and call.dt <= '{{ execution_date | cst_hour }}'
                                   and detail.bill_code = call.waybill_no
                     left join jms_dwd.dwd_tab_barscan_sign_base_hi sign
                               on sign.dt >= '{{ execution_date | hour_add(-1) | cst_hour }}' and sign.dt <= '{{ execution_date | cst_hour }}'
                                   and detail.bill_code = sign.waybill_no
                     left join jms_dm.dm_waybill_customer_complaint_detail_new_dt complaint
                               on complaint.dt >= date_add('{{ execution_date | cst_ds }}', -15) and
                                  complaint.dt <= date_add('{{ execution_date | cst_ds }}', 8)
                                   and complaint.waybill_no = detail.bill_code
                     left join jms_dwd.dwd_tab_barscan_collect_base_hi collect
                               on collect.dt >= '{{ execution_date | hour_add(-1) | cst_hour }}' and collect.dt <= '{{ execution_date | cst_hour }}'
                                   and collect.waybill_no = detail.bill_code
                     left join jms_dm.dm_operate_area_mark_telephone_detail_hi mark
                               on mark.waybill_id = detail.bill_code
                                   and mark.dt >= date_add('{{ execution_date | cst_ds }}', -7)
                                   and mark.dt <= '{{ execution_date | cst_ds }}') detail
      where (is_warning_keys + is_out_violate + is_in_violate +
             is_problem_warning + is_in_call_violate_must + is_problem_time + is_problem_call)
                > 0
      union all
      select *
      from (select nvl(detail.bill_code, leave_detail.waybill_no)                         bill_code,
                   if(detail.deliver_date = '1999-01-01 00:00:00', null, deliver_date)    latest_scan_date,
                   if(detail.deliver_site_code = '', null, deliver_site_code)             latest_scan_code,
                   if(detail.biz_sign_date = '1999-01-01 00:00:00', null, biz_sign_date)  biz_sign_date,
                   if(detail.biz_sign_site_code = '', null, biz_sign_site_code)           biz_sign_site_code,
                   if(detail.deliver_date = '1999-01-01 00:00:00', null,
                      deliver_date)                                                       last_site_come_date,
                   if(detail.deliver_site_name = '', null, deliver_site_name)             last_site_come_name,
                   if(detail.deliver_site_code = '', null, deliver_site_code)             last_site_come_code,
                   if(detail.collect_in_site_date = '1999-01-01 00:00:00', null,
                      collect_in_site_date)                                               collect_in_site_date,
                   if(detail.collect_in_site_name = '', null, collect_in_site_name)       collect_in_site_name,
                   if(detail.collect_in_site_code = '', null, collect_in_site_code)       collect_in_site_code,
                   if(detail.collect_out_site_date = '1999-01-01 00:00:00', null,
                      collect_out_site_date)                                              collect_out_site_date,
                   if(detail.collect_out_site_name = '', null, collect_out_site_name)     collect_out_site_name,
                   if(detail.collect_out_site_code = '', null, collect_out_site_code)     collect_out_site_code,
                   if(detail.receiver_detail_address = '', null, receiver_detail_address) receiver_detail_address,
                   if(detail.problem_desc_code = '', null, detail.problem_desc_code)      problem_desc_code,
                   if(detail.problem_desc = '', null, detail.problem_desc)                problem_desc,
                   0                                                                      is_warning_keys,
                   0                                                                      is_out_violate,
                   0                                                                      is_in_violate,
                   0                                                                      is_in_call_violate_must,
                   0                                                                      is_in_call_violate,
                   0                                                                      is_sign_call_violate,
                   null                                                                   call_time,
                   null                                                                   signed_type,
                   complaint.create_time                                                  create_time,
                   complaint.work_type                                                    work_type,
                   null                                                                   partner_name,
                   null                                                                   storage_address,
                   0                                                                      is_problem,
                   0                                                                      is_problem_warning,
                   0                                                                      is_problem_time,
                   0                                                                      is_problem_call,
                   leave_detail.new_content,
                   leave_Detail.update_time                                               leave_time,
                   if(leave_detail.is_plan = 1 and detail.biz_sign_date is null and collect_out_site_date is null, 1,
                      0)                                                                  is_plan,
                   --  leave_detail.is_leave
                   if(leave_Detail.create_time < nvl(complaint.create_time, '2099-12-31 23:59:59') and
                      leave_Detail.receive_network_code = detail.deliver_site_code, 1, 0) is_leave,
                   if(detail.send_deliver_user_id='' or detail.send_deliver_user_id=0,null,detail.send_deliver_user_id) send_deliver_user_id,
                   if(detail.send_deliver_user='',null,detail.send_deliver_user) send_deliver_user
            from (select leave.waybill_no,
                         leave.new_content,
                         leave.update_time,
                         leave.create_time,
                         leave.receive_network_code,
                         IF(leave.new_content REGEXP '送货上门|距离远', 1, 0)              is_plan,
                         IF(leave.new_content REGEXP
                            '用户未收到|核实签收|签收未收|没找到|没收到|找不到|联系不上业务员|电话无人接听|无人接听|没有收到|催件|没有收到|催派|不要放|未收到|未找到|核实派件|虚假签收',
                            1,
                            0)                                                             is_leave,
                         row_number() over (partition by waybill_no order by create_time ) rn
                  from jms_dwd.dwd_leave_message_hi leave
                  where leave.dt >= '{{ execution_date | hour_add(-1) | cst_hour }}'
                    and leave.dt <= '{{ execution_date | cst_hour }}'
                    and is_delete = 1) leave_detail
                     left join jms_dwd.dwd_wide_yl_rt_ssmx_bill_detail_hf detail
                               on leave_detail.waybill_no = detail.bill_code
                                   and substr(detail.dt, 1, 10) >= date_add('{{ execution_date | cst_ds }}', -7)
                                   and substr(detail.dt, 1, 10) <= '{{ execution_date | cst_ds }}'
                                   and dt <= '{{ execution_date | cst_hour }}'
                     left join jms_dm.dm_waybill_customer_complaint_detail_new_dt complaint
                               on complaint.dt >= date_add('{{ execution_date | cst_ds }}', -15) and
                                  complaint.dt <= date_add('{{ execution_date | cst_ds }}', 8)
                                   and complaint.waybill_no = leave_detail.waybill_no
            where (leave_detail.is_plan = 1
                or leave_detail.is_leave = 1)
              and rn = 1) detail
      where detail.is_plan = 1
         or detail.is_leave = 1) detail  where latest_scan_code is not null
         and date(last_site_come_date)>=date_add('{{ execution_date | cst_ds }}', -7)  and date(last_site_come_date) <= '{{ execution_date | cst_ds }}'
group by bill_code
    distribute by dt
       , 1;

with order as(
  select
    accept_network_code,
    waybill_no,
    create_time,
    '邮政工单' as work_type,
    delivery_name
  from
    jms_dwd.dwd_appeal_info_new_base_hi
  where
    dt >= date_add('{{ execution_date | cst_ds }}', -15)
    and dt <= date_add('{{ execution_date | cst_ds }}', 15)
  union all
  select
    first_duty_network_code,
    waybill_no,
    create_time,
    '理赔工单' as work_type,
    delivery_name
  from
    jms_dwd.dwd_claim_work_order_new_base_hi
  where
    dt >= date_add('{{ execution_date | cst_ds }}', -15)
    and dt <= date_add('{{ execution_date | cst_ds }}', 15)
  union all
  select
    n22.code,
    project.waybill_no,
    project.create_time,
    '平台工单' as work_type,
    staff_name
  from
    jms_dwd.dwd_project_work_order_new_base_hi project
    left join jms_dim.dim_network_whole_massage n22 on project.responsibility_network_id = n22.id
  where
    dt >= date_add('{{ execution_date | cst_ds }}', -15)
    and dt <= date_add('{{ execution_date | cst_ds }}', 15)
  union all
  select
    transfer_network_code,
    waybill_no,
    create_time,
    '普通工单' as work_type,
    delivery_name
  from
    jms_dwd.dwd_work_order_new_base_hi
  where
    dt >= date_add('{{ execution_date | cst_ds }}', -15)
    and dt <= date_add('{{ execution_date | cst_ds }}', 15)
)

insert overwrite table jms_dm.dm_sqs_false_sign_warning_detail_hi partition (dt)
select hours,
       mark_time,
       waybill_no,
       latest_scan_code,
       agent_code,
       virt_code,
       fran_code,
       netowrk_name,
       network_type,
       agent_name,
       virt_name,
       fran_name,
       is_complaint,
       complaint_time,
       work_type,
       mark_reason,
       mark_leavl,
       last_site_come_date,
       collect_in_site_date,
       collect_out_site_date,
       biz_sign_date,
       call_time,
       sender_type,
       sender_type_address,
       signed_type,
       leave_message,
       leave_time,
       problem_second_type,
       problem_second_type_desc,
       is_problem,
       date_time,
       send_deliver_user_id,
       send_deliver_user,
       dt
from (select hour(mark_time)                                     hours,
             mark_time,
             warnning_detail.waybill_no,
             latest_scan_code,
             net.agent_code,
             net.virt_code,
             net.fran_code,
             net.name                                            netowrk_name,
             net.network_type,
             net.agent_name,
             net.virt_name,
             net.fran_name,
             if(complaint.create_time is not null, 1, 0)         is_complaint,
             complaint.create_time                               complaint_time,
             complaint.work_type                                 work_type,
             regexp_replace(reason, '(,@)|(@,)', '')             mark_reason,
             case
                 when all_score = 1 then 1
                 when all_score = 2 then 2
                 when all_score > 2 then 3
                 end                                             mark_leavl,
             last_site_come_date,
             collect_in_site_date,
             collect_out_site_date,
             biz_sign_date,
             call_time,
             sender_type,
             sender_type_address,
             signed_type,
             new_content                                         leave_message,
             leave_time                                          leave_time,
             problem_desc_code                                   problem_second_type,
             warnning_detail.problem_desc                        problem_second_type_desc,
             is_problem,
             date(mark_time)                                     date_time,
             date(mark_time)                                     dt,
             row_number() over (partition by warnning_detail.waybill_no order by if(
                             unix_timestamp(nvl(complaint.create_time, '2099-12-31 23:59:59')) -
                             unix_timestamp(warnning_detail.mark_time) >= 0, 1, 2) ,(
                     unix_timestamp(nvl(complaint.create_time, '2099-12-31 23:59:59')) -
                     unix_timestamp(warnning_detail.mark_time))) rn,
            warnning_detail.send_deliver_user_id,
            nvl(warnning_detail.send_deliver_user,complaint.delivery_name) send_deliver_user
      from (select bill_code                                                 waybill_no,
                   latest_scan_code,
                   is_warning_keys,
                   is_out_violate,
                   is_in_violate,
                   is_in_call_violate,
                   is_sign_call_violate,
                   call_time,
                   create_time                                               complaint_time,
                   work_type                                                 work_type,
                   last_site_come_date,
                   collect_in_site_date,
                   collect_out_site_date,
                   biz_sign_date,
                   partner_name                                              sender_type,
                   storage_address                                           sender_type_address,
                   problem_desc_code,
                   problem_desc,
                   signed_type,
                   is_problem,
                   case
                       when call_time is null then if(is_warning_keys = 1 and collect_in_site_date is not null, 3, 1)
                       when call_time is not null
                           then if(is_warning_keys = 1 and collect_in_site_date is not null, 2, 1)
                       end
                       + if(is_out_violate = 1, 2, 0)
                       + if(is_in_violate = 1, 2, 0)
                       + is_problem_warning
                       + case
                             when call_time is null then if(is_plan = 1 and collect_in_site_date is not null, 3, 1)
                             when call_time is not null then if(is_plan = 1 and collect_in_site_date is not null, 2, 1)
                       end
                       + case
                             when call_time is null then if(is_leave = 1 and collect_in_site_date is not null, 3, 1)
                             when call_time is not null then if(is_leave = 1 and collect_in_site_date is not null, 2, 1)
                       end
                       + if(is_in_call_violate_must = 1, 3, 0)
                       + if(is_problem_time = 1, 1, 0)
                       + if(is_problem_call = 1, 1, 0)
                                                                             all_score,
                   concat_ws(',', map_keys(
                           map(
                                   if(is_warning_keys = 1, '客户有指定派送要求', '@'), is_warning_keys,
                                   if(is_out_violate = 1, '非工作时间出库', '@'), is_out_violate,
                                   if(is_in_violate = 1, '出入库时间间隔小于10s', '@'), is_in_violate,
                                   if(is_problem_warning = 1, '有更址问题件记录', '@'), is_problem_warning,
                                   if(is_plan = 1, '客户有指定派送要求', '@'), is_plan,
                                   if(is_leave = 1, '寄件人反馈客户未收到/找不到包裹', '@'), is_leave,
                                   if(is_in_call_violate_must = 1, '务必电联入库后未电联', '@'),is_in_call_violate_must,
                                   if(is_problem_time = 1, '有更改派送时间问题件记录', '@'), is_problem_time,
                                   if(is_problem_call = 1, '有上报收件客户电话异常问题件记录', '@'), is_problem_call
                               )
                       )
                       )
                                                                             reason,
                   new_content,
                   leave_time,
                   from_unixtime(unix_timestamp(mark_time, 'yyyy-MM-dd-HH')) mark_time,
                   detail.send_deliver_user_id,
                   detail.send_deliver_user
            from (select *,
                         row_number() over (partition by bill_code order by dt desc ) rn,
                         min(dt) over (partition by bill_code order by dt desc )      mark_time
                  from jms_dm.dm_sqs_false_sign_warning_mark_detail_hi  --取最近8天的增量数据，标记时间取最小的时间，运单号取最表标记状态
                  where dt >= date_add('{{ execution_date | cst_ds }}', -7)
                    and dt <= date_add('{{ execution_date | cst_ds }}', 1)
                    and dt <= '{{ execution_date | cst_hour }}') detail
            where rn = 1) warnning_detail
               left join jms_dim.dim_network_whole_massage net
                         on warnning_detail.latest_scan_code = net.code
               left join order complaint --
                         on  complaint.waybill_no = warnning_detail.waybill_no
      where net.network_type = 6
        and date(mark_time) >= date_add('{{ execution_date | cst_ds }}', -7)
        and date(mark_time) <= '{{ execution_date | cst_ds }}') detail
where rn = 1
  and mark_time <= nvl(complaint_time, '2099-12-31 23:59:59')
    distribute by dt
    , abs(hash(waybill_no)) % 5;
